package cn.vetech.sync.elasticsearch.scheduling;

import com.alibaba.otter.canal.client.CanalConnector;
import com.alibaba.otter.canal.protocol.CanalEntry.Entry;
import com.alibaba.otter.canal.protocol.CanalEntry.EntryType;
import com.alibaba.otter.canal.protocol.CanalEntry.EventType;

import cn.vetech.sync.elasticsearch.event.DeleteCanalEvent;
import cn.vetech.sync.elasticsearch.event.InsertCanalEvent;
import cn.vetech.sync.elasticsearch.event.UpdateCanalEvent;

import com.alibaba.otter.canal.protocol.Message;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.BeansException;
import org.springframework.context.ApplicationContext;
import org.springframework.context.ApplicationContextAware;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;
import org.springframework.util.CollectionUtils;

import javax.annotation.Resource;
import java.util.List;

/**
 * @author chenqingsong
 * @version 1.0
 * @since 2018-2-7 17:31:33
 */
@Component
public class CanalScheduling implements Runnable, ApplicationContextAware {
    private static final Logger logger = LoggerFactory.getLogger(CanalScheduling.class);
    private ApplicationContext applicationContext;

    @Resource
    private CanalConnector canalConnector;

    @Scheduled(fixedDelay = 100)
    @Override
    public void run() {
        try {
            int batchSize = 1000;
            // Message message = connector.get(batchSize);
            // 读取canal数据
            Message message = canalConnector.getWithoutAck(batchSize);
            if (null == message) {
                return;
            }
            long batchId = message.getId();
            logger.debug("scheduled_batchId=" + batchId);
            try {
                List<Entry> entries = message.getEntries();
                if (batchId != -1 && !CollectionUtils.isEmpty(entries)) {
                    entries.forEach(entry -> {
                        if (EntryType.ROWDATA == entry.getEntryType()) {
                            publishCanalEvent(entry);
                        }
                    });
                }
                // 通知canal服务端已读数据
                canalConnector.ack(batchId);
            } catch (Exception e) {
                logger.error("发送监听事件失败！batchId回滚,batchId=" + batchId, e);
                canalConnector.rollback(batchId);
            }
        } catch (Exception e) {
            logger.error("canal_scheduled异常！", e);
        }
    }

    /**
     * 将改变的数据推送给具体的监听器
     *
     * @param entry Entry 数据库改变数据
     */
    private void publishCanalEvent(Entry entry) {
        EventType eventType = entry.getHeader().getEventType();
        switch (eventType) {
            case INSERT:
                applicationContext.publishEvent(new InsertCanalEvent(entry));
                break;
            case UPDATE:
                applicationContext.publishEvent(new UpdateCanalEvent(entry));
                break;
            case DELETE:
                applicationContext.publishEvent(new DeleteCanalEvent(entry));
                break;
            default:
                break;
        }
    }

    @Override
    public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {
        this.applicationContext = applicationContext;
    }
}
